package com.yeeiee.utils;

import com.yeeiee.catalog.ClinkCatalog;
import com.yeeiee.factory.ClinkFactory;
import com.yeeiee.sink.ClinkSink;
import com.yeeiee.source.ClinkSource;
import com.yeeiee.transform.ClinkTransform;
import lombok.val;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.util.Preconditions;

import java.util.ServiceLoader;

/**
 * 工厂 工具类
 */
public final class FactoryUtils {

    /**
     * 根据标识符获取 source
     */
    public static ClinkSource getSource(String identifier, Configuration configuration) {
        return getIdentifier(ClinkSource.class, identifier, configuration);
    }

    /**
     * 根据标识符获取 sink
     */
    public static ClinkSink getSink(String identifier, Configuration configuration) {
        return getIdentifier(ClinkSink.class, identifier, configuration);
    }

    /**
     * 根据标识符获取 transform
     */
    public static ClinkTransform getTransform(String identifier, Configuration configuration) {
        return getIdentifier(ClinkTransform.class, identifier, configuration);
    }

    /**
     * 根据标识符获取 catalog
     */
    public static ClinkCatalog getCatalog(String identifier, Configuration configuration) {
        return getIdentifier(ClinkCatalog.class, identifier, configuration);
    }

    private static <T extends ClinkFactory<T>> T getIdentifier(Class<T> clazz, String identifier, Configuration configuration) {
        Preconditions.checkNotNull(identifier, "identifier cannot be null");
        Preconditions.checkNotNull(configuration, "configuration cannot be null");

        val factories = ServiceLoader.load(clazz);
        for (T factory : factories) {
            if (identifier.equalsIgnoreCase(factory.factoryIdentifier())) {
                factory.validate(configuration);
                return factory.build(configuration);
            }
        }

        throw new CatalogException("Unable to find identifier: " + identifier);
    }
}
